package com.shujia.youhua

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

//使用reduceByKey/aggregateByKey替代groupByKey
object AggregateByKeyDemo {
  def main(args: Array[String]): Unit = {
    val ss: SparkSession = SparkSession.builder()
      .master("local")
      .appName("缓存，避免使用重复的RDD")
      .getOrCreate()
    val sc: SparkContext = ss.sparkContext

    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    val stuArrRDD: RDD[(String, String, Int, String, String)] = lineRDD.map(_.split(","))
      .map {
        case Array(id: String, name: String, age: String, gender: String, clazz: String) => (id, name, age.toInt, gender, clazz)
      }

    //计算每个班级的人数

    /**
     * 使用groupBy实现,任意一种RDD都能调用
     * 只做分组，不做聚合操作。
     */
//    val resRDD: RDD[(String, Int)] = stuArrRDD.groupBy(_._5)
//      .map((kv: (String, Iterable[(String, String, Int, String, String)])) => {
//        (kv._1, kv._2.toList.size)
//      })
//    resRDD.foreach(println)


    val stuKvRDD: RDD[(String, Int)] = stuArrRDD.map {
      case (id: String, name: String, age: Int, gender: String, clazz: String) => (clazz, 1)
    }

    /**
     * reduceByKey，只能是KV类的RDD才能调用
     * 将前一个RDD中的分区数据分到后一个RDD中不同的分区中，会产生shuffle
     * 在后一个RDD中的分区内部进行聚合，聚合的方式，一点一点的聚合
     */
//    val resRDD2: RDD[(String, Int)] = stuKvRDD.reduceByKey(_ + _)
//    resRDD2.foreach(println)

    /**
     * aggregateByKey，只能是KV类的RDD才能调用
     * 真正意义上先在前一个RDD分区中聚合一次，然后在后一个RDD分区中再聚合
     * 自己自定义map端【前一个RDD分区】聚合逻辑
     * 自己自定义reduce端【后一个RDD分区】聚合逻辑
     */
    //aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,combOp: (U, U) => U)
    /**
     * zeroValue: 初始值，只会被前一个RDD预聚合开始进行使用
     * seqOp: 自定义map端【前一个RDD分区】聚合逻辑
     * combOp: 自定义reduce端【后一个RDD分区】聚合逻辑
     */
    val resRDD: RDD[(String, Int)] = stuKvRDD.aggregateByKey(0)(
      (a1: Int, a2: Int) => a1 + a2,
      (b1: Int, b2: Int) => b1 + b2,
    )

    resRDD.foreach(println)



  }
}
